package com.shujia.kafka

import java.{lang, util}
import java.util.Properties

import org.apache.kafka.clients.consumer.{ConsumerRecord, ConsumerRecords, KafkaConsumer}

object Demo3Consumer {
  def main(args: Array[String]): Unit = {

    /**
      * 创建消费者
      *
      */

    val properties = new Properties()

    //1、kafka broker列表
    properties.setProperty("bootstrap.servers", "master:9092,node1:9092,node2:9092")

    //2、指定消费者组
    properties.setProperty("group.id", "asdasdad")


    //key vvalue 反序列化的类
    properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")


    /**
      * earliest
      * 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
      * latest
      * 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
      * none
      * topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常
      *
      */


    //从最早读取数据
    properties.put("auto.offset.reset", "earliest")


    val consumer = new KafkaConsumer[String, String](properties)


    /**
      * 订阅topic
      *
      */

    val topics = new util.ArrayList[String]()

    topics.add("student3")

    consumer.subscribe(topics)


    /**
      *
      * 消费数据
      */
    while (true) {

      println("消费数据")


      //从kafka中拉取数据，一次拉多条数据
      val records: ConsumerRecords[String, String] = consumer.poll(1000)

      //解析数据,获取指定topic的数据
      val iterable: lang.Iterable[ConsumerRecord[String, String]] = records.records("student3")

      //返回一个迭代器
      val iter: util.Iterator[ConsumerRecord[String, String]] = iterable.iterator()

      while (iter.hasNext) {
        //一行数据
        val record: ConsumerRecord[String, String] = iter.next()

        val key: String = record.key()
        val topic: String = record.topic()
        val partition: Int = record.partition()
        val ts: Long = record.timestamp()
        val value: String = record.value()

        println(s"$key\t$topic\t$partition\t$ts\t$value")
      }
    }


  }

}
